package com.idanchuang.support.etcache.pubsub.amqp;

import com.idanchuang.support.etcache.pubsub.AbstractRemoteEventPublisher;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 基于 AMQP 实现的缓存变更事件发布
 *
 * @author yjy
 * Created at 2021/12/22 11:12 上午
 */
public class AmqpRemoteEventPublisher extends AbstractRemoteEventPublisher {

    private static final Logger logger = LoggerFactory.getLogger(AmqpRemoteEventPublisher.class);

    private final ConnectionFactory connectionFactory;
    private final String exchange;
    private Connection connection;
    private Channel channel;

    public AmqpRemoteEventPublisher(String appId, ConnectionFactory connectionFactory) {
        try {
            this.connectionFactory = connectionFactory;
            this.exchange = "etcache-event-" + appId.toLowerCase();
            this.connection = connectionFactory.newConnection();
            this.channel = connection.createChannel();
            this.channel.exchangeDeclare(this.exchange, "fanout");
        } catch (Exception e) {
            throw new RuntimeException("AmqpRemoteEventPublisher init failed", e);
        }
    }

    @Override
    public void doPublish(byte[] remoteKeyBytes) throws Exception {
        if (!this.channel.isOpen()) {
            reconnect();
        }
        AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder().build();
        this.channel.basicPublish(this.exchange, "", properties, remoteKeyBytes);
    }

    /**
     * 重连amqp
     */
    private void reconnect() {
        while (!isShutdown() && !this.channel.isOpen()) {
            try {
                Thread.sleep(1000L);
                if (!this.connection.isOpen()) {
                    this.connection = this.connectionFactory.newConnection();
                }
                this.channel = this.connection.createChannel();
                logger.info("reconnect amqp success, new channelNumber: {}", this.channel.getChannelNumber());
            } catch (InterruptedException e) {
                logger.warn("reconnect amqp cancel, cause thread is interrupted", e);
                break;
            } catch (Exception e) {
                logger.error("reconnect amqp failed, will retry after 1s..", e);
            }
        }
    }

    public ConnectionFactory getConnectionFactory() {
        return connectionFactory;
    }

    public Connection getConnection() {
        return connection;
    }

    public Channel getChannel() {
        return channel;
    }

    @Override
    public void destroy() {
        try {
            super.destroy();
            if (this.connection != null && this.connection.isOpen()) {
                this.connection.close();
            }
            logger.info("AmqpRemoteEventSubscriber destroy finish > exchange: {}", exchange);
        } catch (Exception e) {
            logger.warn("connection close failed, cause: {}", e.getMessage());
        }
    }
}
